package com.vip.kafka.controller;

import com.vip.kafka.response.RestResponse;
import com.vip.kafka.response.RestResponses;
import com.vip.kafka.service.IKafkaService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;

import java.util.List;

/**
 * kafka消息推送 控制器
 *
 * @author wgb
 * @date 2020/5/19 14:07
 */
@Slf4j
@RestController
@RequestMapping("/api/kafka")
public class KafkaController {

    @Autowired
    private IKafkaService kafkaService;

    /**
     * 创建主题
     *
     * @param topic
     * @return
     */
    @PostMapping("/topic")
    public RestResponse<String> createTopic(@RequestParam String topic){
        return RestResponses.newResponseFromResult(kafkaService.createTopic(topic));
    }

    /**
     * 删除主题
     *
     * @param topic
     * @return
     */
    @DeleteMapping("/topic/{topic}")
    public RestResponse<String> deleteTopic(@PathVariable String topic){
        return RestResponses.newResponseFromResult(kafkaService.deleteTopic(topic));
    }

    /**
     * 主题列表
     *
     * @return
     */
    @GetMapping("/topic/list")
    public RestResponse<List<String>> getTopicList(){
        return RestResponses.newResponseFromResult(kafkaService.topicList());
    }


    /**
     * 消息推送到Kafka服务
     *
     * @param topic
     * @param message
     */
    @PostMapping("/send/{topic}")
    public RestResponse sendMessage(@PathVariable String topic, @RequestBody String message){
        return RestResponses.newResponseFromResult(kafkaService.sendMessage(topic, message));
    }

}
